package lhb.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;


import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.util.Properties;

import org.apache.kafka.clients.producer.ProducerRecord;

public class kafkaproducer {

    public static void main(String[] args) {

        //创建kafka配置参数
        Properties properties = new Properties();
        //kafka集群
        properties.put("bootstrap.servers", "192.168.0.201:9092,192.168.0.202:9092,192.168.0.203:9092");
        //重试次数
        properties.put("acks", "all");
        //重试次数
        properties.put("retries", 1);
        //批次大小
        properties.put("batch.size", 16384);
        //等待时间
        properties.put("linger.ms", 1);
        //RecordAccumulator 缓冲区大小
        properties.put("buffer.memory", 33554432);
        properties.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");


        KafkaProducer<String, String> producerconfig = new KafkaProducer<>(properties);
        //通过字节流读文件数据
        try {
            BufferedReader bf = new BufferedReader(
                    new FileReader(
                            new File(
                                    "F:\\FIXALARMS1.txt")));
            String line = null;
            while((line=bf.readLine())!=null){
                producerconfig.send(new ProducerRecord<>("sstream", line), (metadata, exception) -> {
                    if(exception == null){
                        System.out.println(metadata.partition()+"--"+metadata.serializedValueSize()+"-"+metadata.toString());
                    }
                });
                /*
                想通过KeyedMessage，没找到jar包，告辞
                KeyedMessage<String,String> keyedMessage = new KeyedMessage<String,String>("JsonData3",line);
                Thread.sleep(5000);
                producer.send(keyedMessage);*/
            }
            bf.close();
            System.out.println("已经发送完毕");

        } catch (Exception e) {
            e.printStackTrace();
        }


        producerconfig.close();




    }
}
